package com.atguigu.upp.app;

import com.atguigu.upp.bean.TagInfo;
import com.atguigu.upp.bean.TaskInfo;
import com.atguigu.upp.bean.TaskTagRule;
import com.atguigu.upp.bean.TypeCodeConstant;
import com.atguigu.upp.service.MysqlDBService;
import com.atguigu.upp.utils.PropertiesUtil;
import com.atguigu.upp.utils.UPPUtil;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2022/11/11
 *
 *      目前获取的输入:
 *              taskid: 任务的id
 *              doDate： 基于哪一天的数仓中的数据计算
 */
public class TaskExecuteApp
{
    public static void main(String[] args) throws IOException {


        //接收参数
        /*String taskId = args[0];
        String doDate = args[1]; */

        String taskId = args[0];
        String doDate = args[1];

        //准备编程环境
        SqlSessionFactory sqlSessionFactory = UPPUtil.getSqlSessionFactoryByConfig("config.xml");
        MysqlDBService mysqlDBService = new MysqlDBService(sqlSessionFactory.openSession());
        SparkSession sparkSession = UPPUtil.getSparkSession("TaskExecuteApp");

        //根据taskId去查询计算的任务的相关信息
        TaskInfo taskInfo = mysqlDBService.getTaskInfoByTaskId(taskId);
        TagInfo tagInfo = mysqlDBService.getTagInfoByTaskId(taskId);
        List<TaskTagRule> taskTagRules = mysqlDBService.getTaskTagRulesByTaskId(taskId);

        //获取语句
        String cretaeTableSql = generateCretaeTableSql(tagInfo, taskInfo);
        String insertSql = generateInsertSql(tagInfo, taskInfo, taskTagRules,doDate);


       //计算sql
       //sparkSession.sql(cretaeTableSql);
       //sparkSession.sql(insertSql);

        //------------------



    }

    /*
    insert overwrite table upp220704.TAG_POPULATION_ATTRIBUTES_NATURE_GENDER partition (dt='2020-06-14')
    select
        uid,
        --只有有四级标签的三级标签才需要封装
        case tagValue
            when 'U' then '未知'
            when 'F' then '男性'
            when 'M' then '女性'
        end tagValue
    from
    (
           task_sql
    ) tmp;
     */
    private static String generateInsertSql(TagInfo tagInfo, TaskInfo taskInfo,List<TaskTagRule> taskTagRules,String doDate){

        String template = "insert overwrite table %s.%s partition (dt='%s') "+
                          "select uid, %s from ( %s ) tmp";


        //获取库名表名
        String dbName = PropertiesUtil.getProperty("updbname");
        String tableName = tagInfo.getTagCode();

        //处理taskSql
        String taskSql = taskInfo.getTaskSql().replace(';', ' ').replace("currentDate",doDate);

        String caseStr = "";
        //确定case部分 只有有四级标签的三级标签才需要封装
        if (taskTagRules.size() > 0){
            //有规则
            String str = taskTagRules
                .stream()
                .map(t -> String.format("  when '%s' then '%s' ", t.getQueryValue(), t.getSubTagValue()))
                .collect(Collectors.joining(" "));

            caseStr = "case tagValue " + str + " end tagValue ";

        }else{
            //没有四级标签的封装规则
            caseStr = "tagValue";
        }

        String sql = String.format(template, dbName, tableName, doDate, caseStr, taskSql);

        System.out.println(sql);

        return sql;

    }

    private static String generateCretaeTableSql(TagInfo tagInfo, TaskInfo taskInfo){

        String template = "create table if not exists %s.%s(uid string,tagValue %s ) " +
                            "comment '%s' " +
                            "partitioned by (dt string) " +
                            "row format delimited fields terminated by '\\t' " +
                            "location '%s/%s' ";


        //获取库名
        String dbName = PropertiesUtil.getProperty("updbname");
        String tableName = tagInfo.getTagCode();
        String tableComment = taskInfo.getTaskComment();
        String hdfsPath = PropertiesUtil.getProperty("hdfsPath");

        //获取当前标签计算值的类型
        String tagValueType ="";

        switch (tagInfo.getTagValueType()){
            case TypeCodeConstant.TAG_VALUE_TYPE_LONG: tagValueType = "bigint" ; break;
            case TypeCodeConstant.TAG_VALUE_TYPE_DECIMAL: tagValueType = "decimal(16,0)" ; break;
            case TypeCodeConstant.TAG_VALUE_TYPE_STRING: tagValueType = "string" ; break;
            case TypeCodeConstant.TAG_VALUE_TYPE_DATE: tagValueType = "string" ; break;
        }

        //格式化
        String sql = String.format(template, dbName, tableName, tagValueType, tableComment, hdfsPath + dbName, tableName);

        System.out.println(sql);

        return sql;


    }
}
